/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.neptune_test.common;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.nhncorp.neptune.client.NTable;
import com.nhncorp.neptune.client.NTableManager;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.aop.TestHandler;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.NTimeoutException;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.common.util.NetworkUtil;
import com.nhncorp.neptune.master.NeptuneMaster;
import com.nhncorp.neptune.tablet.ColumnValue;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tabletserver.ColumnCollection;
import com.nhncorp.neptune.tabletserver.DataServiceProtocol;
import com.nhncorp.neptune.tabletserver.Tablet;
import com.nhncorp.neptune.tabletserver.TabletManagerProtocol;
import com.nhncorp.neptune.tabletserver.TabletServerIF;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.protocol.LockMetaData;

public abstract class NeptuneClusterTestCase extends NeptuneTestCase {
	final int MAX_READY_TIME = 180; //3 min

	protected MiniNeptuneCluster cluster;
	protected final boolean format;
	protected int numMasterServers = 1;
	protected int numTabletServers;

	protected Map<String, List<String>> getHandlers() {
		return new HashMap<String, List<String>>();
	}

	protected NeptuneClusterTestCase(String name, final boolean format) {
		this(name, format, 1);
	}

	protected NeptuneClusterTestCase(String name, final boolean format,
			final int numTabletServers) {
		this(name, format, 1, numTabletServers);
	}

	protected NeptuneClusterTestCase(String name, final boolean format,
			final int numMasterServers, final int numTabletServers) {
		super(name);
		this.format = format;
		this.numMasterServers = numMasterServers;
		this.numTabletServers = numTabletServers;
		//100MB
		conf.set("maxTabletSize", 100);
		conf.set("tabletServer.majorCompactionTime", 100 * 60); //majorCompaction 주기를 100분으로 설정
	}

	@Override
	public void setUp() throws Exception {
		super.setUp();
		this.cluster = new MiniNeptuneCluster(this.conf, this.numMasterServers,
				this.numTabletServers, this.format, getHandlers());
		waitClusterInit();
	}

	@Override
	public void tearDown() throws Exception {
		System.out
				.println("======================TearDown================================");
		super.tearDown();
		if (this.cluster != null) {
			this.cluster.shutdown();
		}
		try {
			Thread.sleep(15 * 1000);
		} catch (InterruptedException e) {
		}
	}

	protected void waitClusterInit() throws IOException {
		int count = 0;
		while (true) {
			if (NeptuneMaster.isClusterReady()) {
				break;
			}
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			count++;
			if (count > MAX_READY_TIME) {
				fail("exceeds MAX_READY_TIME");
			}
		}
	}

	protected Object[] findTabletInfo(String tableName) throws IOException {
		long startTime = System.currentTimeMillis();
		int timeout = 1 * 60 * 1000; //1min
		Map<String, TabletServerIF> tabletServers = cluster.getTabletServers();
		TabletServerIF tabletServer = null;
		TabletInfo tablet = null;
		for (Map.Entry<String, TabletServerIF> entry : tabletServers.entrySet()) {
			TabletServerIF eachTabletServer = entry.getValue();
			TabletInfo[] tabletInfos = eachTabletServer.reportTablets();
			if (tabletInfos == null) {
				continue;
			}
			for (TabletInfo eachTabletInfo : tabletInfos) {
				if (tableName.equals(eachTabletInfo.getTableName())) {
					tabletServer = eachTabletServer;
					tablet = eachTabletInfo;
					break;
				}
			}
			if ((System.currentTimeMillis() - startTime) > timeout) {
				throw new NTimeoutException(
						"Timeout call NeptuneClusterTestCase.findTabletInfo: "
								+ tableName);
			}
		}
		return new Object[] { tabletServer, tablet };
	}

	/**
	 * Tablet 정보가 META에 저장될때 까지 기다린다.
	 * MasterServer fail이 특정 table이 META에 저장된 다음에 처리해야 할 경우 사용한다.
	 * @param table
	 * @throws IOException
	 */
	protected void waitUntilSaveToMetaTablet(NTable ntable) throws IOException {
		long startTime = System.currentTimeMillis();
		int timeout = 1 * 60 * 1000; //1min

		boolean end = false;
		while (end) {
			try {
				TabletInfo[] tabletInfos = ntable.listTabletInfos();
				if (tabletInfos == null || tabletInfos.length == 0) {
					continue;
				}

				Object[] metaInfo = findTabletInfo(Constants.TABLE_NAME_META);
				if (metaInfo == null || metaInfo.length == 0) {
					continue;
				}
				TabletServerIF metaTabletServer = (TabletServerIF) metaInfo[0];
				TabletInfo metaTablet = (TabletInfo) metaInfo[1];
				Tablet metaTable = metaTabletServer.getTablet(metaTablet
						.getTabletName());
				ColumnCollection columnCollection = metaTable
						.getMemorySSTable().getColumnCollections().get(
								Constants.META_COLUMN_NAME_TABLETINFO);
				if (columnCollection == null) {
					continue;
				}

				end = false;
				for (TabletInfo eachTabletInfo : tabletInfos) {
					Row.Key rowKey = Tablet.generateMetaRowKey(eachTabletInfo
							.getTableName(), eachTabletInfo.getEndRowKey());
					ColumnValue[] columnValues = columnCollection.get(rowKey);
					if (columnValues == null || columnValues.length == 0) {
						continue;
					}
					if (!eachTabletInfo.getAssignedHostName().equals(
							columnValues[0].getValueAsString())) {
						end = false;
					} else {
						end = true;
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
			if ((System.currentTimeMillis() - startTime) > timeout) {
				throw new NTimeoutException(
						"Timeout call NeptuneClusterTestCase.waitTabletAssign: "
								+ ntable.descTable());
			}
		}
	}

	protected List<TabletInfo> listTabletInfos(String tableName) {
		List<TabletInfo> result = new ArrayList<TabletInfo>();
		Map<String, TabletServerIF> tabletServers = cluster.getTabletServers();
		TabletServerIF tabletServer = null;
		TabletInfo tablet = null;
		for (Map.Entry<String, TabletServerIF> entry : tabletServers.entrySet()) {
			TabletServerIF eachTabletServer = entry.getValue();
			TabletInfo[] tabletInfos = eachTabletServer.reportTablets();
			if (tabletInfos == null) {
				continue;
			}
			for (TabletInfo eachTabletInfo : tabletInfos) {
				if (tableName.equals(eachTabletInfo.getTableName())) {
					result.add(eachTabletInfo);
				}
			}
		}
		return result;
	}

	protected TabletManagerProtocol connectTabletManager(
			String tabletServerHostName, NConfiguration conf)
			throws IOException {
		return (TabletManagerProtocol) NRPC.getProxy(
				TabletManagerProtocol.class, TabletManagerProtocol.versionID,
				NetworkUtil.getAddress(tabletServerHostName), conf);
	}

	protected DataServiceProtocol connectTabletServer(TabletInfo tabletInfo,
			NConfiguration conf) throws IOException {
		if (tabletInfo == null) {
			return null;
		}
		return connectTabletServer(tabletInfo.getAssignedHostName(), conf);
	}

	protected static DataServiceProtocol connectTabletServer(String hostName,
			NConfiguration conf) throws IOException {
		if (hostName == null) {
			return null;
		}
		DataServiceProtocol tabletServer = (DataServiceProtocol) NRPC.getProxy(
				DataServiceProtocol.class, DataServiceProtocol.versionID,
				NetworkUtil.getAddress(hostName), conf);

		return tabletServer;
	}

	protected void registerTestEvent(TestTarget testTarget) throws IOException,
			Exception, InterruptedException {
		String lockPath = TestHandler.TEST_LOCK_PATH + "/"
				+ testTarget.testMethodName;
		//targetClass, methodName, className, before
		String contentKey = testTarget.testTargetClass.getName() + ","
				+ testTarget.testMethodName + ","
				+ testTarget.testProcessorClass.getName() + ","
				+ testTarget.before + "," + testTarget.onlyOne;
		LockMetaData lockMetaData = new LockMetaData();
		lockMetaData.setLockId(lockPath);
		lockMetaData.setContentKey(contentKey);

		if (!NTableManager.getLockService(conf).createNode(lockMetaData, true)) {
			throw new Exception("Can't lock for test:" + lockPath);
		}
		Thread.sleep(2000);
		NTableManager.getLockService(conf).releaseLock(lockPath,
				LockService.LockMode.LOCK_EXCLUSIVE);
	}

	protected List<String> getListTabletServer(LockService lockService)
			throws Exception {
		List<String> tabletServers = new ArrayList<String>();

		Map<String, LockMetaData> servers = lockService
				.readDir(Constants.SERVER);

		for (LockMetaData lockMetaData : servers.values()) {
			String lockId = lockMetaData.getLockId();
			String hostName = NetworkUtil.fileNameToHostName(lockId
					.substring(Constants.SERVER.length() + 1));
			if (!lockService.tryLock(lockId,
					LockService.LockMode.LOCK_EXCLUSIVE)) {
				try {
					DataServiceProtocol tabletServer = connectTabletServer(
							hostName, conf);
					tabletServer.test(0, "");
					//에러가 없는 경우 add
					tabletServers.add(hostName);
				} catch (Exception e) {
				}
			}
		}
		return tabletServers;
	}

	protected class TestTarget {
		Class testTargetClass;
		String testMethodName;
		Class testProcessorClass;
		boolean before;
		boolean onlyOne;

		public TestTarget(Class testTargetClass, String testMethodName,
				Class testProcessorClass, boolean before) {
			this(testTargetClass, testMethodName, testProcessorClass, before,
					true);
		}

		public TestTarget(Class testTargetClass, String testMethodName,
				Class testProcessorClass, boolean before, boolean onlyOne) {
			this.testTargetClass = testTargetClass;
			this.testMethodName = testMethodName;
			this.testProcessorClass = testProcessorClass;
			this.before = before;
			this.onlyOne = onlyOne;
		}
	}
}
